package com.github.facadefs.ftp;

import com.github.facadefs.FileView;
import com.github.facadefs.support.SessionBasedFileSystem;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.*;
import org.apache.hadoop.fs.ftp.FTPException;
import com.github.facadefs.Selector;
import com.github.facadefs.util.Paths;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.StringTokenizer;

import static com.github.facadefs.util.Paths.PATH_SEPARATOR;

/**
 * FTP FileSystem.
 *
 * @author vacoor
 */
@Slf4j
@Getter
@Setter
public class FtpFileSystem extends SessionBasedFileSystem<FTPClient> {
    public static final int DEFAULT_PORT = FTP.DEFAULT_PORT;

    private static final String DEFAULT_CONTROL_ENCODING = StandardCharsets.UTF_8.name();
    private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;

    private String host;
    private int port;
    private String username;
    private String password;

    private String chroot;

    /**
     * {@inheritDoc}
     */
    @Override
    protected FTPClient connect() throws IOException {
        final String host = this.host;
        final int port = 0 < this.port ? this.port : DEFAULT_PORT;
        final String username = this.username;
        final String password = this.password;

        final FTPClient client = new FTPClient();
        client.connect(host, port);
        client.enterLocalPassiveMode();

        final int reply = client.getReplyCode();
        if (!FTPReply.isPositiveCompletion(reply)) {
            throw new IOException(String.format("Server - %s refused connection on port - %s", host, port));
        } else if (client.login(username, password)) {
            client.setFileType(FTP.BINARY_FILE_TYPE);
            client.setBufferSize(DEFAULT_BUFFER_SIZE);
            client.setFileTransferMode(FTP.BLOCK_TRANSFER_MODE);
            client.setControlEncoding(DEFAULT_CONTROL_ENCODING);
        } else {
            throw new IOException(String.format("Login failed on server - %s, port - %s", host, port));
        }
        return client;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void disconnect(FTPClient ftp) {
        if (null == ftp) {
            return;
        }
        try {
            try {
                if (ftp.isConnected() && !ftp.logout()) {
                    log.warn("Logout failed while disconnect, error code - " + ftp.getReplyCode());
                }
            } finally {
                ftp.disconnect();
            }
        } catch (final IOException ex) {
            log.warn("Server - {}:{} disconnect error: {}", host, port, ex.getMessage());
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected FileView stat(final FTPClient ftp, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        return stat0(ftp, absPath);
    }

    private FileView stat0(final FTPClient ftp, final String absPath) throws IOException {
        if (null == absPath || Paths.ROOT.equals(absPath)) {
            return new FileView(Paths.ROOT, -1, true, 0, 0, 0, null, null, null);
        }

        final int i = absPath.lastIndexOf(PATH_SEPARATOR);
        final String absParentPath = i <= 0 ? null : absPath.substring(0, i);
        final String targetName = i < 0 ? absPath : absPath.substring(i + 1);

        final FTPFile[] candidates = ftp.listFiles(absParentPath, new FTPFileFilter() {
            @Override
            public boolean accept(final FTPFile candidate) {
                return null != candidate && targetName.equals(candidate.getName());
            }
        });
        return 0 < candidates.length ? toView(ftp, absPath, candidates[0]) : null;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected FileView[] ls(final FTPClient ftp, final String path) throws IOException {
        final String absPath = toAbsolutePath(path);
        final FileView stat = stat0(ftp, absPath);

        if (null == stat) {
            return new FileView[0];
        }
        if (!stat.isDirectory()) {
            return new FileView[]{stat};
        }

        final FTPFile[] children = ftp.listFiles(absPath);
        final FileView[] stats = new FileView[children.length];
        for (int i = 0; i < children.length; i++) {
            stats[i] = toView(ftp, absPath, children[i]);
        }
        return stats;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void ls(final FTPClient ftp, final String path, final Selector selector) throws IOException {
        final String absPath = toAbsolutePath(path);
        final FileView stat = stat0(ftp, absPath);

        if (null == stat) {
            return;
        }
        if (!stat.isDirectory()) {
            selector.select(stat);
            return;
        }
        final FTPFile[] children = ftp.listFiles(absPath);
        for (final FTPFile child : children) {
            final FileView v = toView(ftp, absPath, child);
            if (Selector.CONTINUE != selector.select(v)) {
                break;
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected boolean mkdirs(final FTPClient ftp, final String path) throws IOException {
        final StringTokenizer tokenizer = new StringTokenizer(path, PATH_SEPARATOR, false);

        String parentPath = Paths.ROOT;
        while (tokenizer.hasMoreTokens()) {
            final String name = tokenizer.nextToken();
            final String pathToUse = Paths.ROOT.equals(parentPath) ? parentPath + name : parentPath + PATH_SEPARATOR + name;
            final String absParentPath = toAbsolutePath(parentPath);
            final FileView stat = stat0(ftp, toAbsolutePath(pathToUse));
            if (null == stat) {
                if (!ftp.changeWorkingDirectory(absParentPath)) {
                    throw new IOException("Can't change working directory to " + absParentPath);
                }
                if (!ftp.makeDirectory(name)) {
                    return false;
                }
            } else if (!stat.isDirectory()) {
                throw new IOException("File already exists and is not directory: " + pathToUse);
            }
            parentPath = pathToUse;
        }
        return true;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void rm(final FTPClient ftp, final String path, final boolean recurse) throws IOException {
        final String absPath = toAbsolutePath(path);
        final FileView stat = stat0(ftp, absPath);

        if (null == stat) {
            return;
        }
        if (stat.isDirectory()) {
            final FTPFile[] children = ftp.listFiles(absPath);
            if (!recurse && 0 < children.length) {
                throw new IOException("You cannot delete non-empty directory, use recurse=true to overide");
            }
            for (final FTPFile child : children) {
                String name = child.getName();
                if (Paths.isCurrentDirName(name) || Paths.isParentDirName(name)) {
                    continue;
                }
                rm(ftp, toAbsolutePath(absPath, child.getName()), recurse);
            }
            if (!ftp.removeDirectory(absPath)) {
                throw new IOException("cannot delete directory: " + absPath);
            }
        } else {
            if (!ftp.deleteFile(absPath)) {
                throw new IOException("cannot delete file: " + absPath);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void rename(FTPClient ftp, String src, String dst) throws IOException {
        final String absSrc = toAbsolutePath(src);
        final String absDst = toAbsolutePath(dst);

        if (null == stat0(ftp, absSrc)) {
            throw new FileNotFoundException("Source path " + src + " does not exist");
        }
        if (null != stat0(ftp, absDst)) {
            throw new IOException("Destination path " + dst + " already exist, cannot rename!");
        }

        // FIXME 可能是目录重命名
        if (!ftp.rename(absSrc, absDst)) {
            throw new IOException("rename failed: " + src + " --> " + dst);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public InputStream open(final FTPClient ftp, final String path) throws IOException {
        return open(ftp, path, DEFAULT_BUFFER_SIZE);
    }

    protected InputStream open(final FTPClient ftp, final String path, final int bufferSize) throws IOException {
        final String absPat = toAbsolutePath(path);
        final FileView stat = stat0(ftp, absPat);
        if (null == stat) {
            return null;
        }
        if (stat.isDirectory()) {
            throw new IOException("Can't open " + path + " because it is a directory");
        }

        ftp.allocate(bufferSize);

        final int i = absPat.lastIndexOf(PATH_SEPARATOR);
        final String absParentPath = i < 0 ? Paths.ROOT : absPat.substring(0, i + 1);
        final String name = i < 0 ? absPat : absPat.substring(i + 1);
        // Change to parent directory on the
        // server. Only then can we read the
        // file
        // on the server by opening up an InputStream. As a side effect the working
        // directory on the server is changed to the parent directory of the file.
        // The FTP client connection is closed when close() is called on the
        // FSDataInputStream.
        if (!ftp.changeWorkingDirectory(absParentPath)) {
            throw new IOException("Can't change working directory to " + absParentPath);
        }

        final InputStream in = new DataInputStream(ftp.retrieveFileStream(name)) {
            @Override
            public void close() throws IOException {
                super.close();
                completePendingCommand(ftp);
            }
        };
        if (!FTPReply.isPositivePreliminary(ftp.getReplyCode())) {
            // The ftpClient is an inconsistent state. Must close the stream
            // which in turn will logout and disconnect from FTP server
            throw new IOException("Unable to open file: " + path + ", Aborting");
        }
        return in;
    }

    private void completePendingCommand(final FTPClient ftp) throws IOException {
        if (!ftp.isConnected()) {
            throw new FTPException("Client not connected");
        } else {
            final boolean cmdCompleted = ftp.completePendingCommand();
            if (!cmdCompleted) {
                throw new FTPException("Could not complete transfer, Reply Code - " + ftp.getReplyCode());
            }
        }
    }

    @Override
    protected OutputStream write(final FTPClient ftp, final String path, final String contentType, final boolean override) throws IOException {
        return write(ftp, path, override, DEFAULT_BUFFER_SIZE);
    }

    private OutputStream write(final FTPClient ftp, final String path, final boolean override, int bufferSize) throws IOException {
        final String absPath = toAbsolutePath(path);
        final int i = absPath.lastIndexOf(PATH_SEPARATOR);
        final String name = i < 0 ? absPath : absPath.substring(i + 1);
        final String absParentPath = i < 0 ? Paths.ROOT : absPath.substring(0, i + 1);

        final FileView stat = stat0(ftp, absPath);
        if (null != stat) {
            if (!override) {
                throw new IOException("File already exists: " + path);
            } else if (stat.isDirectory()) {
                throw new IOException("Can't override " + path + " because it is a directory");
            }
        } else if (!this.mkdirs(ftp, absParentPath)) {
            throw new IOException("mkdirs failed to create: " + absParentPath);
        }

        ftp.allocate(bufferSize);
        // Change to parent directory on the server. Only then can we write to the
        // file on the server by opening up an OutputStream. As a side effect the
        // working directory on the server is changed to the parent directory of the
        // file. The FTP client connection is closed when close() is called on the
        // FSDataOutputStream.
        ftp.changeWorkingDirectory(absParentPath);
        final OutputStream out = new DataOutputStream(ftp.storeFileStream(name)) {
            @Override
            public void close() throws IOException {
                super.close();
                completePendingCommand(ftp);
            }
        };
        if (!FTPReply.isPositivePreliminary(ftp.getReplyCode())) {
            // The ftpClient is an inconsistent state. Must close the stream
            // which in turn will logout and disconnect from FTP server
            throw new IOException("Unable to create file: " + path + ", Aborting");
        }
        return out;
    }

    private FileView toView(FTPClient ftp, String absPath, FTPFile ftpFile) {
        long length = ftpFile.getSize();
        boolean isDir = ftpFile.isDirectory();
        long lastModifiedTime = ftpFile.getTimestamp().getTimeInMillis();
        long lastAccessTime = 0;
        int permissions = getPermissions(ftpFile);
        String user = ftpFile.getUser();
        String group = ftpFile.getGroup();
        String link = ftpFile.getLink();
        String path = toRelativePath(absPath);

        return new FileView(path, length, isDir, lastModifiedTime, lastAccessTime, permissions, user, group, link);
    }

    private int getPermissions(FTPFile ftpFile) {
        int permissions = 0;
        if (ftpFile.hasPermission(FTPFile.USER_ACCESS, FTPFile.READ_PERMISSION)) {
            permissions |= FileView.OWNER_READ;
        }
        if (ftpFile.hasPermission(FTPFile.USER_ACCESS, FTPFile.WRITE_PERMISSION)) {
            permissions |= FileView.OWNER_WRITE;
        }
        if (ftpFile.hasPermission(FTPFile.USER_ACCESS, FTPFile.EXECUTE_PERMISSION)) {
            permissions |= FileView.OWNER_EXECUTE;
        }

        if (ftpFile.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.READ_PERMISSION)) {
            permissions |= FileView.GROUP_READ;
        }
        if (ftpFile.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.WRITE_PERMISSION)) {
            permissions |= FileView.GROUP_WRITE;
        }
        if (ftpFile.hasPermission(FTPFile.GROUP_ACCESS, FTPFile.EXECUTE_PERMISSION)) {
            permissions |= FileView.GROUP_EXECUTE;
        }

        if (ftpFile.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.READ_PERMISSION)) {
            permissions |= FileView.OTHER_READ;
        }
        if (ftpFile.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.WRITE_PERMISSION)) {
            permissions |= FileView.OTHER_WRITE;
        }
        if (ftpFile.hasPermission(FTPFile.WORLD_ACCESS, FTPFile.EXECUTE_PERMISSION)) {
            permissions |= FileView.OTHER_EXECUTE;
        }
        return permissions;
    }

    /* *********************************************
     *
     * *********************************************/

    private String toAbsolutePath(final String path) throws IOException {
        return Paths.toAbsolutePath(chroot, path);
    }

    private String toAbsolutePath(final String base, final String path) throws IOException {
        return Paths.toAbsolutePath(base, path);
    }

    private String toRelativePath(final String path) {
        return Paths.toRelativePath(chroot, path);
    }

}
